package name.yzhu.wbdata.weibo;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import name.yzhu.common.EnvVariable;
import name.yzhu.common.concurrent.ConcurrentException;
import name.yzhu.common.concurrent.Executable;
import name.yzhu.common.concurrent.TaskLoopManager;
import name.yzhu.common.conf.PropertyConfig;
import name.yzhu.common.io.IOUtil;
import name.yzhu.common.service.IServer;
import name.yzhu.common.service.StopBase;
import name.yzhu.common.service.Stoppable;
import name.yzhu.common.util.IdentifyUnique;
import name.yzhu.common.yaml.YamlItem;
import name.yzhu.httpserver.ServerException;
import name.yzhu.wbdata.io.IPersistWeibo;
import name.yzhu.wbdata.io.PersistIdentifyUnique;
import name.yzhu.wbdata.io.PersistYaml;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class WeiboService implements IServer {
	private static final Log LOG = LogFactory.getLog(WeiboService.class);
	private static final String OUTPUT_FILE = "output.file";
	private static final String STOP_WEIBO = "stop.weibo";
	private static final String DEFAULT_OUTPUT_FILE = EnvVariable.getTempDirectory()+File.separator+"weibo.yaml";
	protected Stoppable stop = null;
	private Stoppable stopOutput = null;
	protected IPersistWeibo yaml = null;
	private IdentifyUnique identifyUnique = null;
	
	protected PropertyConfig cfg = null;

	private List<TaskLoopManager> weibos = new ArrayList<TaskLoopManager>();
	private BlockingQueue<YamlItem> queue = new ArrayBlockingQueue<YamlItem>(3000);
	private TaskLoopManager outputManager = null;
	protected volatile boolean running = false;
	private volatile boolean joining = false;

	public WeiboService(Stoppable s, PropertyConfig cfg) {
		stop = s;
		this.cfg  = cfg;
	}

	public void init() throws WeiboException {
		try {
			String ignoredWeibo = cfg.getString(STOP_WEIBO);
			Pattern p = null;
			if (null != ignoredWeibo)
			   p = Pattern.compile(ignoredWeibo);
			for (WeiboType wb : WeiboType.values()) {
				if (null != p && p.matcher(wb.getType()).find()) {
					continue;
				}
				AbstractWeiboSDK weibo = WeiboSDKFactory.getWeiboSDK(wb, queue);
				weibo.open(cfg);
				TaskLoopManager manage = new TaskLoopManager(1000, stop, weibo);
				manage.setData(wb.getType());
				weibos.add(manage);
				LOG.info("weibo type:"+wb.getType()+" task No: "+ manage.getTaskNo());
			}
			
			initPersist();
			
			stopOutput = new StopBase();
			OutputTask outputTask = new OutputTask(queue, yaml) ;
			outputManager  = new TaskLoopManager(500, stopOutput, outputTask);
			LOG.info("output thread No: "+ outputManager.getTaskNo());
		} catch (WeiboException e) {
			LOG.error("initialize weibo server failed ", e);
			throw new WeiboException(e);
		}

	}
	
	protected void initPersist() {
		String outputFile =  cfg.getString(OUTPUT_FILE,
				DEFAULT_OUTPUT_FILE);
		if (!IOUtil.isAbsolutePath(outputFile)) 
			outputFile = EnvVariable.getWorkingDirectory()+File.separator
			             +outputFile;
		identifyUnique = new PersistIdentifyUnique(outputFile);
		yaml = new PersistYaml(outputFile, identifyUnique);
	}

	@Override
	public void start() throws ServerException {
		try {
			if (isRunning())
				return;
			
			yaml.open();
			if (null != identifyUnique)
			  identifyUnique.open();
			for (TaskLoopManager wb : weibos){
				try {
					wb.start();
				} catch (Exception e) {
					LOG.error("start weibo sdk failed "+(String)wb.getData(), e);
				}
			}
			outputManager.start();
			running = true;
			LOG.info("weibo server is running ...");
		} catch (IOException e) {
			LOG.error("start weibo server failed ", e);
			throw new ServerException(e);
		} catch (Exception e) {
			LOG.error("start weibo server failed ", e);
			throw new ServerException(e);
		}
	}

	@Override
	public void join() throws ServerException {
		if (joining)
			return;
		joining = true;
		LOG.info("weibo server is joining ...");
		try {
			for (TaskLoopManager  m: weibos) {
				m.join();
			}
			stopOutput.stop();
			outputManager.join();
			if (null != yaml){
				yaml.close();
				yaml = null;
			}
			if (null != identifyUnique){
				identifyUnique.close();
				identifyUnique = null;
			}

		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
		finally{
			joining = false;
		    LOG.info("weibo server is joined ...");
	    }
	}

	@Override
	public boolean isJoining() {
		return joining  ;
	}
	
	@Override
	public void stop() throws ServerException {
		try{
			if (!isRunning())
				return ;
			LOG.info("weibo server is stopping ...");
			while(!stop.isStopped()){
				Thread.sleep(500);
			}
		    yaml.close();
		    if (null != identifyUnique)
		    	identifyUnique.close();
		    running = false;
		} catch (IOException e) {
			LOG.error(e.getMessage(), e);
		} catch (InterruptedException e) {
			LOG.error(e.getMessage(), e);
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}
		finally{
		    LOG.info("weibo server is stopped ...");
		}
	}

	static class OutputTask implements Executable {
		private IPersistWeibo yaml = null;
		private BlockingQueue<YamlItem> queue = null;

		public OutputTask(BlockingQueue<YamlItem> queue, IPersistWeibo persist) {
			super();
			this.queue = queue;
			yaml = persist;
		}

		@Override
		public void execute() throws ConcurrentException {
			try {
				YamlItem doc;
				while((doc = queue.poll(100, TimeUnit.MILLISECONDS)) != null){
					try {
						yaml.persistYaml(doc);
					} catch (Exception e) {
						LOG.error("persist yaml filed yaml id  "
								+doc.getLong(name.yzhu.wbdata.Constant.YAML_FIELD_WEIBO_ID), e);
					}
				}

			} catch (Exception e) {
				LOG.error(e.getMessage(), e);
			}
		}
	}

	@Override
	public boolean isRunning() throws Exception {
		return running ;
	}
}
